package com.xxx.rpc.client;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.xxx.rpc.common.bean.RpcRequest;
import com.xxx.rpc.common.bean.RpcResponse;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;

/**
 * 通过wait 和 notify 机制，将异步的netty NIO通信过程转换成同步的调用；
 * @author Rosun
 *
 */
public class DefauldCallBack implements CallBack{

	private static final Logger LOGGER = LoggerFactory.getLogger(DefauldCallBack.class);
	//private Object lock = new Object();
	
	private RpcResponse resp = null;
	
	//private transient boolean called = false;
	private CountDownLatch latch = new CountDownLatch(1);
	
	
	@Override
	public void doCallBack(RpcResponse resp) {
		LOGGER.debug("begin doCallBack " + resp.toString());
		this.resp = resp;
		this.latch.countDown();
//		synchronized (this.lock) {
//			this.called = true;
//			lock.notifyAll();
//		}
		LOGGER.debug("end doCallBack " + resp.toString());
	}

	@Override
	public RpcResponse get(final String requestId,long timeout) {
		try {
			boolean zero = this.latch.await(timeout,TimeUnit.MILLISECONDS);
			if(zero){
				return this.resp;
			}else{
				//false if the waiting time elapsed before the count reached zero
				this.resp = new RpcResponse();
				resp.setException(new TimeoutException("time out, in ["+ timeout + "] MILLISECONDS"));
				resp.setRequestId(requestId);
			}
		} catch (InterruptedException e) {
			this.resp = new RpcResponse();
			this.resp.setException(e);
			this.resp.setRequestId(requestId);
		}
		return this.resp;
	}
	
	@Override
	public RpcResponse get(final String requestId) {
		try {
			this.latch.await();
		} catch (InterruptedException e) {
			this.resp = new RpcResponse();
			this.resp.setException(e);
			this.resp.setRequestId(requestId);
		}
		return this.resp;
	}

	@Override
	public void destroy() {
		//this.lock = null;
		this.resp = null;
	}

	@Override
	public void doSendThroughChannel(Channel channel, RpcRequest request) throws IOException{
		LOGGER.debug("begin 2 doSendThroughChannel " + request.toString());
		try {
			ChannelFuture cf =  channel.writeAndFlush(request);
			if(cf != null){
				cf.sync(); //直到消息发送出去 才返回！
			}
		} catch (Exception e) {
			//如果有任何异常： channel
			throw new IOException(e);
		}
		LOGGER.debug("end 2 doSendThroughChannel " + request.toString());
	}

}
